package com.zhike.app.ods;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.zhike.utils.KafkaUtils;
import com.zhike.utils.MysqlUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;



/**
 * @Author: zhike
 * @CreateTime: 2024/1/17
 * @Description:  通过flink cdc实时监测mysql中的维度数据并同步到Kafka
 */
public class OdsApp {

    public static void main(String[] args) {

        //1.获取执行环境
        Configuration conf = new Configuration();
        //设置web端口
//        conf.setInteger("rest.port",10000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        //2.通过flink cdc 读取mysql中的维度数据并创建流
        MySqlSource<String> mySqlSource = MysqlUtils.getMysqlSource("car_data.car_info");
        DataStreamSource<String> streamSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "cdc-source");

        //3.维度数据同步到kafka
        KafkaSink<String> kafkaSink = KafkaUtils.getKafkaSink("car_db");

        streamSource.print();
        streamSource.sinkTo(kafkaSink);
        //4.执行任务
        try {
            env.execute("ods_cdc");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}
